package com.sunday.canal.adapter.mongo.adapter;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.sunday.canal.adapter.mongo.constants.LogConstants;
import com.sunday.canal.adapter.mongo.support.DateTimeUtil;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;

@Slf4j
public class BaseMongoStarter {

    protected volatile boolean running = false;
    protected Thread.UncaughtExceptionHandler handler = (_, e) -> log.error("parse events has an error, {}", e.toString(), e);
    protected Thread thread = null;
    @Setter
    protected CanalConnector connector;
    protected String destination;

    protected void printSummary(Message message, long batchId, int size) {
        long memsize = 0;
        for (CanalEntry.Entry entry : message.getEntries()) {
            memsize += entry.getHeader().getEventLength();
        }

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntries())) {
            startPosition = buildPositionForDump(message.getEntries().get(0));
            endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
        }

        log.info(LogConstants.context_format, batchId, size, memsize, DateTimeUtil.format(), startPosition,
                endPosition);
    }

    protected String buildPositionForDump(CanalEntry.Entry entry) {
        String position = STR."\{entry.getHeader().getLogfileName()}:\{entry.getHeader().getLogfileOffset()}:\{entry.getHeader().getExecuteTime()}(\{DateTimeUtil.format(entry.getHeader().getExecuteTime())})";
        if (StringUtils.isNotEmpty(entry.getHeader().getGtid())) {
            position += STR." gtid(\{entry.getHeader().getGtid()})";
        }
        return position;
    }

}
